package org.datastreamapi.rich

import org.FlinkStreamApp
import org.apache.flink.api.common.functions.RichMapFunction
import org.apache.flink.configuration.Configuration
import org.apache.flink.streaming.api.scala._

/**
 * description ：MapFunction 富函数
 * author      ：剧情再美终是戏 
 * mail        : 13286520398@163.com
 * date        ：Created in 2020/2/21 20:29
 * modified By ：
 * version:    : 1.0
 */
object RichMapFunction extends FlinkStreamApp {

  override def doSomeThing(environment: StreamExecutionEnvironment) = {
    // 创建数据
    val source = environment.fromElements(1, 2, 3)

    // 转换 RichMapFunction RichFlatMapFunction RichFilterFunction
    source.map(new MyRichMapFunction).print
  }

  class MyRichMapFunction extends RichMapFunction[Int, Int] {

    override def open(parameters: Configuration): Unit = {
      println("进入map函数了！")
    }

    override def map(value: Int): Int = value + 1

    override def close(): Unit = {
      println("结束map函数了！")
    }
  }

}
